package com.cellulam.framework.mq.send;

import com.aliyun.openservices.ons.api.Admin;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.cellulam.framework.core.utils.JacksonUtils;
import com.cellulam.framework.core.base.ResourceId;
import com.cellulam.framework.core.spring.SmartEnvironment;
import com.cellulam.framework.mq.RocketMqConfig;
import com.cellulam.properties.CellulamConstans;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.Properties;

@Slf4j
public abstract class BaseAbstractMessageSender<P extends Admin>
        implements MessageSender {

    private RocketMqConfig mqConfig;

    private String groupId;
    private String topic;

    private P producer;

    protected BaseAbstractMessageSender(String groupId, String topic) {
        this.groupId = SmartEnvironment.env.resolveIfNeed(groupId);
        this.topic = SmartEnvironment.env.resolveIfNeed(topic);

        this.mqConfig = this.initMqConfig();
        Properties properties = this.mqConfig.toMqProperties();
        if (this.getSendTimeoutMillis() != null) {
            properties.put(PropertyKeyConst.SendMsgTimeoutMillis, this.getSendTimeoutMillis());
        }
        this.producer = this.createProducer(properties);
        this.producer.start();
    }

    private RocketMqConfig initMqConfig() {
        String accessKey = SmartEnvironment.env.getRequiredProperty("ali.mq.access-key");
        String secretKey = SmartEnvironment.env.getRequiredProperty("ali.mq.secret-key");
        String nameSrv = SmartEnvironment.env.getRequiredProperty("ali.mq.name-srv");

        return RocketMqConfig.builder()
                .accessKey(accessKey)
                .secretKey(secretKey)
                .nameSrv(nameSrv)
                .sendTimeoutMillis(this.getSendTimeoutMillis())
                .groupId(this.groupId)
                .build();
    }


    protected Integer getSendTimeoutMillis() {
        return 1000;
    }

    /**
     * 延时消息
     * 需要确认topic是否为定时/延时topic
     *
     * @param tag
     * @param body
     * @param delayMillis
     * @return
     */
    public <T extends ResourceId> SendResult send(String tag, T body, long delayMillis) {
        return this.sendAtDeliverTime(tag, body, System.currentTimeMillis() + delayMillis);
    }

    /**
     * 定时消息
     * 需要确认topic是否为定时/延时topic
     *
     * @param tag
     * @param body
     * @param startDeliverTime
     * @return
     */
    public <T extends ResourceId> SendResult send(String tag, T body, LocalDateTime startDeliverTime) {
        Long time = startDeliverTime.toInstant(CellulamConstans.DEFAULT_ZONE_OFFSET).toEpochMilli();
        return this.sendAtDeliverTime(tag, body, time);
    }

    private <T extends ResourceId> SendResult sendAtDeliverTime(String tag, T body, Long startDeliverTime) {
        String resourceId = body.getResourceId() == null ? null : body.getResourceId().toString();
        Message message = new Message(this.topic,
                tag,
                resourceId,
                this.convertToBody(body));
        message.setShardingKey(resourceId);

        if (startDeliverTime != null) {
            message.setStartDeliverTime(startDeliverTime);
        }

        SendResult sendResult = this.send(message);

        log.info("Send MQ message, group: {}, tag: {}, resourceId: {} message: {}, startDeliverTime: {}",
                this.mqConfig.getGroupId(),
                tag,
                resourceId,
                sendResult,
                startDeliverTime);
        return sendResult;
    }

    /**
     * 立即发送
     *
     * @param tag
     * @param body
     * @return
     */
    public final <T extends ResourceId> SendResult send(String tag, T body) {
        return this.sendAtDeliverTime(tag, body, null);
    }

    protected <T extends ResourceId> byte[] convertToBody(T body) {
        return JacksonUtils.toJson(body).getBytes();
    }

    protected P getProducer() {
        return this.producer;
    }

    protected abstract P createProducer(Properties properties);

    protected abstract SendResult send(Message message);

    protected final String getGroupId() {
        return this.groupId;
    }

    protected final String getTopic() {
        return this.topic;
    }
}
